# -*- coding: utf-8 -*-
"""
Python 企业级推导式和集合操作最佳实践示例

本模块演示了Python推导式和集合操作的企业级最佳实践，包括:
- 列表、字典、集合推导式
- 生成器表达式和内存优化
- 嵌套推导式和复杂条件
- 函数式编程模式
- 性能优化技巧
- 数据转换和过滤
- 并行处理和异步操作
- 错误处理和类型安全
"""

import asyncio
import functools
import itertools
import logging
import math
import operator
import time
from collections import Counter, defaultdict, namedtuple
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import (
    Any, Callable, Dict, Generator, Iterable, Iterator, List, Optional, 
    Set, Tuple, TypeVar, Union, Hashable
)
import sys

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(levelname)s:%(name)s:%(message)s'
)
logger = logging.getLogger(__name__)

# 类型变量
T = TypeVar('T')
U = TypeVar('U')
K = TypeVar('K', bound=Hashable)
V = TypeVar('V')

class ComprehensionType(Enum):
    """推导式类型枚举"""
    LIST = auto()
    DICT = auto()
    SET = auto()
    GENERATOR = auto()
    NESTED = auto()
    CONDITIONAL = auto()

@dataclass
class PerformanceMetrics:
    """性能指标"""
    operation_name: str
    execution_time: float = 0.0
    memory_usage: int = 0
    items_processed: int = 0
    items_per_second: float = 0.0
    memory_per_item: float = 0.0

class ComprehensionProfiler:
    """推导式性能分析器"""
    
    def __init__(self):
        self.metrics: List[PerformanceMetrics] = []
    
    def profile_operation(self, operation_name: str):
        """装饰器：分析操作性能"""
        def decorator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.perf_counter()
                start_memory = sys.getsizeof(args) + sys.getsizeof(kwargs)
                
                try:
                    result = func(*args, **kwargs)
                    
                    end_time = time.perf_counter()
                    end_memory = sys.getsizeof(result)
                    
                    execution_time = end_time - start_time
                    memory_usage = end_memory - start_memory
                    
                    # 计算处理的项目数量
                    items_processed = 0
                    if hasattr(result, '__len__'):
                        try:
                            items_processed = len(result)
                        except (TypeError, AttributeError):
                            items_processed = 1
                    elif hasattr(result, '__iter__'):
                        items_processed = 1
                    
                    # 创建性能指标
                    metrics = PerformanceMetrics(
                        operation_name=operation_name,
                        execution_time=execution_time,
                        memory_usage=memory_usage,
                        items_processed=items_processed,
                        items_per_second=items_processed / execution_time if execution_time > 0 else 0,
                        memory_per_item=memory_usage / items_processed if items_processed > 0 else 0
                    )
                    
                    self.metrics.append(metrics)
                    return result
                    
                except Exception as e:
                    logger.error(f"操作 '{operation_name}' 失败: {e}")
                    raise
            
            return wrapper
        return decorator
    
    def get_metrics(self) -> List[PerformanceMetrics]:
        """获取所有性能指标"""
        return self.metrics.copy()
    
    def get_summary(self) -> Dict[str, Any]:
        """获取性能摘要"""
        if not self.metrics:
            return {}
        
        total_time = sum(m.execution_time for m in self.metrics)
        total_memory = sum(m.memory_usage for m in self.metrics)
        total_items = sum(m.items_processed for m in self.metrics)
        
        return {
            'total_operations': len(self.metrics),
            'total_time': total_time,
            'total_memory': total_memory,
            'total_items': total_items,
            'avg_time_per_operation': total_time / len(self.metrics),
            'avg_items_per_second': total_items / total_time if total_time > 0 else 0
        }

# 全局性能分析器
profiler = ComprehensionProfiler()

# 测试数据
Student = namedtuple('Student', ['name', 'age', 'grade', 'subjects'])
Product = namedtuple('Product', ['id', 'name', 'price', 'category', 'in_stock'])

def generate_test_data() -> Tuple[List[Student], List[Product], List[int], List[str]]:
    """生成测试数据"""
    students = [
        Student("Alice", 20, 85, ["Math", "Physics", "Chemistry"]),
        Student("Bob", 19, 92, ["Math", "Computer Science"]),
        Student("Charlie", 21, 78, ["Physics", "Chemistry", "Biology"]),
        Student("Diana", 20, 95, ["Math", "Physics", "Computer Science"]),
        Student("Eve", 22, 88, ["Chemistry", "Biology"]),
        Student("Frank", 19, 76, ["Math", "Biology"]),
        Student("Grace", 21, 91, ["Physics", "Computer Science", "Math"]),
        Student("Henry", 20, 83, ["Chemistry", "Physics"]),
    ]
    
    products = [
        Product(1, "Laptop", 999.99, "Electronics", True),
        Product(2, "Mouse", 29.99, "Electronics", True),
        Product(3, "Keyboard", 79.99, "Electronics", False),
        Product(4, "Monitor", 299.99, "Electronics", True),
        Product(5, "Chair", 199.99, "Furniture", True),
        Product(6, "Desk", 399.99, "Furniture", False),
        Product(7, "Book", 19.99, "Education", True),
        Product(8, "Pen", 2.99, "Education", True),
        Product(9, "Notebook", 5.99, "Education", True),
        Product(10, "Headphones", 149.99, "Electronics", True),
    ]
    
    numbers = list(range(1, 101))
    words = [
        "python", "programming", "comprehension", "list", "dictionary", 
        "set", "generator", "function", "class", "module", "package", 
        "import", "variable", "loop", "condition", "exception", "file", 
        "data", "structure", "algorithm"
    ]
    
    return students, products, numbers, words

@profiler.profile_operation("基础列表推导式")
def demonstrate_basic_list_comprehensions() -> None:
    """演示基础列表推导式"""
    print("\n======== 1. 基础列表推导式 ========")
    
    _, _, numbers, words = generate_test_data()
    
    # 基础推导式
    squares = [x**2 for x in range(10)]
    print(f"✓ 平方数: {squares}")
    
    # 条件过滤
    evens = [x for x in numbers if x % 2 == 0]
    print(f"✓ 偶数 (前10个): {evens[:10]}")
    
    # 字符串操作
    upper_words = [word.upper() for word in words if len(word) > 5]
    print(f"✓ 长单词大写: {upper_words}")
    
    # 数学运算
    math_operations = [x**2 + 2*x + 1 for x in range(1, 11)]
    print(f"✓ 数学运算 (x²+2x+1): {math_operations}")
    
    # 类型转换
    string_numbers = [str(x) for x in range(1, 11)]
    print(f"✓ 数字转字符串: {string_numbers}")

@profiler.profile_operation("字典推导式")
def demonstrate_dict_comprehensions() -> None:
    """演示字典推导式"""
    print("\n======== 2. 字典推导式 ========")
    
    students, products, numbers, words = generate_test_data()
    
    # 基础字典推导式
    squares_dict = {x: x**2 for x in range(1, 11)}
    print(f"✓ 平方数字典: {dict(list(squares_dict.items())[:5])}...")
    
    # 学生成绩字典
    student_grades = {student.name: student.grade for student in students}
    print(f"✓ 学生成绩: {student_grades}")
    
    # 产品价格字典（有库存的）
    available_prices = {product.name: product.price for product in products if product.in_stock}
    print(f"✓ 有库存产品价格: {available_prices}")
    
    # 单词长度字典
    word_lengths = {word: len(word) for word in words}
    print(f"✓ 单词长度: {dict(list(word_lengths.items())[:5])}...")
    
    # 条件字典推导式
    high_performers = {student.name: student.grade for student in students if student.grade >= 90}
    print(f"✓ 高分学生: {high_performers}")
    
    # 分组字典
    products_by_category = {}
    for category in set(product.category for product in products):
        products_by_category[category] = [product.name for product in products if product.category == category]
    print(f"✓ 按类别分组: {products_by_category}")

@profiler.profile_operation("集合推导式")
def demonstrate_set_comprehensions() -> None:
    """演示集合推导式"""
    print("\n======== 3. 集合推导式 ========")
    
    students, products, numbers, words = generate_test_data()
    
    # 基础集合推导式
    unique_remainders = {x % 7 for x in numbers}
    print(f"✓ 除7余数集合: {unique_remainders}")
    
    # 学生年龄集合
    student_ages = {student.age for student in students}
    print(f"✓ 学生年龄集合: {student_ages}")
    
    # 产品类别集合
    product_categories = {product.category for product in products}
    print(f"✓ 产品类别: {product_categories}")
    
    # 所有学科集合
    all_subjects = {subject for student in students for subject in student.subjects}
    print(f"✓ 所有学科: {all_subjects}")
    
    # 首字母集合
    first_letters = {word[0].upper() for word in words if word}
    print(f"✓ 单词首字母: {first_letters}")
    
    # 条件集合
    expensive_categories = {product.category for product in products if product.price > 100}
    print(f"✓ 高价产品类别: {expensive_categories}")

@profiler.profile_operation("生成器表达式")
def demonstrate_generator_expressions() -> None:
    """演示生成器表达式"""
    print("\n======== 4. 生成器表达式 ========")
    
    _, _, numbers, words = generate_test_data()
    
    # 基础生成器表达式
    squares_gen = (x**2 for x in range(10))
    print(f"✓ 平方数生成器: {list(squares_gen)}")
    
    # 内存效率比较
    print("\n✓ 内存效率比较:")
    
    # 列表推导式
    large_list = [x**2 for x in range(10000)]
    list_memory = sys.getsizeof(large_list)
    print(f"  列表推导式内存: {list_memory:,} 字节")
    
    # 生成器表达式
    large_gen = (x**2 for x in range(10000))
    gen_memory = sys.getsizeof(large_gen)
    print(f"  生成器表达式内存: {gen_memory:,} 字节")
    print(f"  内存节省: {((list_memory - gen_memory) / list_memory * 100):.1f}%")
    
    # 惰性求值演示
    print("\n✓ 惰性求值演示:")
    def expensive_operation(x):
        print(f"    处理 {x}")
        return x**3
    
    # 生成器不会立即执行
    expensive_gen = (expensive_operation(x) for x in range(5))
    print("  生成器已创建，但未执行操作")
    
    # 只有在迭代时才执行
    print("  开始迭代:")
    first_three = list(itertools.islice(expensive_gen, 3))
    print(f"  前三个结果: {first_three}")

@profiler.profile_operation("嵌套推导式")
def demonstrate_nested_comprehensions() -> None:
    """演示嵌套推导式"""
    print("\n======== 5. 嵌套推导式 ========")
    
    students, _, _, _ = generate_test_data()
    
    # 矩阵操作
    matrix = [[i*j for j in range(1, 4)] for i in range(1, 4)]
    print(f"✓ 乘法表矩阵: {matrix}")
    
    # 扁平化嵌套列表
    nested_list = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
    flattened = [item for sublist in nested_list for item in sublist]
    print(f"✓ 扁平化列表: {flattened}")
    
    # 学生-学科组合
    student_subject_pairs = [(student.name, subject) 
                            for student in students 
                            for subject in student.subjects]
    print(f"✓ 学生-学科对 (前5个): {student_subject_pairs[:5]}")
    
    # 条件嵌套推导式
    high_grade_subjects = [subject 
                          for student in students 
                          if student.grade >= 85
                          for subject in student.subjects]
    subject_count = Counter(high_grade_subjects)
    print(f"✓ 高分学生学科统计: {dict(subject_count)}")
    
    # 复杂嵌套字典
    student_subject_grades = {
        student.name: {
            subject: student.grade 
            for subject in student.subjects
        }
        for student in students
    }
    print(f"✓ 学生学科成绩 (Alice): {student_subject_grades.get('Alice', {})}")

@profiler.profile_operation("条件推导式")
def demonstrate_conditional_comprehensions() -> None:
    """演示条件推导式"""
    print("\n======== 6. 条件推导式 ========")
    
    students, products, numbers, words = generate_test_data()
    
    # 三元运算符
    grade_categories = ["优秀" if student.grade >= 90 
                       else "良好" if student.grade >= 80 
                       else "及格" if student.grade >= 60 
                       else "不及格" 
                       for student in students]
    print(f"✓ 成绩分类: {grade_categories}")
    
    # 复杂条件过滤
    math_physics_students = [student.name for student in students 
                            if "Math" in student.subjects and "Physics" in student.subjects]
    print(f"✓ 数学物理双修学生: {math_physics_students}")
    
    # 价格区间分类
    price_ranges = {
        product.name: "昂贵" if product.price > 200 
                     else "中等" if product.price > 50 
                     else "便宜"
        for product in products
    }
    print(f"✓ 产品价格分类: {price_ranges}")
    
    # 多条件组合
    premium_available_products = [
        product.name for product in products 
        if product.price > 100 and product.in_stock and product.category == "Electronics"
    ]
    print(f"✓ 高端有库存电子产品: {premium_available_products}")
    
    # 字符串条件处理
    processed_words = [word.upper() if len(word) > 6 
                      else word.capitalize() if len(word) > 3 
                      else word.lower() 
                      for word in words]
    print(f"✓ 条件字符串处理 (前10个): {processed_words[:10]}")

@profiler.profile_operation("函数式推导式")
def demonstrate_functional_comprehensions() -> None:
    """演示函数式推导式"""
    print("\n======== 7. 函数式推导式 ========")
    
    _, _, numbers, words = generate_test_data()
    
    # 使用内置函数
    word_info = [(word, len(word), word.count('a')) for word in words]
    print(f"✓ 单词信息 (前5个): {word_info[:5]}")
    
    # 使用lambda函数
    squared_evens = list(map(lambda x: x**2, filter(lambda x: x % 2 == 0, numbers[:20])))
    comprehension_squared_evens = [x**2 for x in numbers[:20] if x % 2 == 0]
    print(f"✓ 偶数平方 (map/filter): {squared_evens[:5]}")
    print(f"✓ 偶数平方 (推导式): {comprehension_squared_evens[:5]}")
    
    # 函数组合
    def is_prime(n):
        if n < 2:
            return False
        return all(n % i != 0 for i in range(2, int(n**0.5) + 1))
    
    def fibonacci(n):
        a, b = 0, 1
        for _ in range(n):
            a, b = b, a + b
        return a
    
    prime_fibs = [fibonacci(i) for i in range(20) if is_prime(fibonacci(i))]
    print(f"✓ 质数斐波那契数: {prime_fibs}")
    
    # 高阶函数应用
    operations = [lambda x: x**2, lambda x: x**3, lambda x: x*2, lambda x: x+10]
    results = {f"op_{i}": [op(x) for x in range(1, 6)] for i, op in enumerate(operations)}
    print(f"✓ 多种运算结果: {results}")

async def async_comprehension_demo(items: List[Any], 
                                  processor: Callable[[Any], Any]) -> List[Any]:
    """异步推导式演示"""
    async def process_item(item):
        if asyncio.iscoroutinefunction(processor):
            return await processor(item)
        else:
            await asyncio.sleep(0.001)  # 模拟异步操作
            return processor(item)
    
    tasks = [process_item(item) for item in items]
    return await asyncio.gather(*tasks)

@profiler.profile_operation("异步推导式")
async def demonstrate_async_comprehensions() -> None:
    """演示异步推导式"""
    print("\n======== 8. 异步推导式 ========")
    
    # 异步处理函数
    async def async_square(x):
        await asyncio.sleep(0.01)
        return x**2
    
    # 测试数据
    test_numbers = list(range(1, 11))
    
    print(f"✓ 处理数据: {test_numbers}")
    
    # 异步处理
    start_time = time.perf_counter()
    async_results = await async_comprehension_demo(test_numbers, async_square)
    async_time = time.perf_counter() - start_time
    
    print(f"✓ 异步结果: {async_results}")
    print(f"✓ 异步处理耗时: {async_time:.4f}秒")
    
    # 同步对比
    start_time = time.perf_counter()
    sync_results = [x**2 for x in test_numbers]
    sync_time = time.perf_counter() - start_time
    
    print(f"✓ 同步结果: {sync_results}")
    print(f"✓ 同步处理耗时: {sync_time:.6f}秒")

async def run_async_tests() -> None:
    """运行异步测试"""
    await demonstrate_async_comprehensions()

def demonstrate_performance_optimization() -> None:
    """演示性能优化技巧"""
    print("\n======== 9. 性能优化技巧 ========")
    
    # 大数据集
    large_data = list(range(100000))
    
    # 测试不同方法的性能
    print("✓ 性能对比测试:")
    
    # 方法1: 列表推导式
    start_time = time.perf_counter()
    result1 = [x**2 for x in large_data if x % 2 == 0]
    time1 = time.perf_counter() - start_time
    print(f"  列表推导式: {time1:.4f}秒, 结果数量: {len(result1)}")
    
    # 方法2: filter + map
    start_time = time.perf_counter()
    result2 = list(map(lambda x: x**2, filter(lambda x: x % 2 == 0, large_data)))
    time2 = time.perf_counter() - start_time
    print(f"  filter+map: {time2:.4f}秒, 结果数量: {len(result2)}")
    
    # 方法3: 传统循环
    start_time = time.perf_counter()
    result3 = []
    for x in large_data:
        if x % 2 == 0:
            result3.append(x**2)
    time3 = time.perf_counter() - start_time
    print(f"  传统循环: {time3:.4f}秒, 结果数量: {len(result3)}")
    
    # 方法4: 生成器（内存优化）
    start_time = time.perf_counter()
    result4_gen = (x**2 for x in large_data if x % 2 == 0)
    # 只计算前1000个
    result4 = list(itertools.islice(result4_gen, 1000))
    time4 = time.perf_counter() - start_time
    print(f"  生成器(前1000): {time4:.6f}秒, 结果数量: {len(result4)}")
    
    # 内存使用比较
    list_memory = sys.getsizeof(result1)
    gen_memory = sys.getsizeof((x**2 for x in large_data if x % 2 == 0))
    print(f"\n✓ 内存使用对比:")
    print(f"  列表内存: {list_memory:,} 字节")
    print(f"  生成器内存: {gen_memory:,} 字节")
    print(f"  内存节省: {((list_memory - gen_memory) / list_memory * 100):.1f}%")

def run_comprehensive_tests() -> None:
    """运行综合测试套件"""
    print("\n======== 10. 综合测试套件 ========")
    
    test_functions = [
        ("基础列表推导式测试", demonstrate_basic_list_comprehensions),
        ("字典推导式测试", demonstrate_dict_comprehensions),
        ("集合推导式测试", demonstrate_set_comprehensions),
        ("生成器表达式测试", demonstrate_generator_expressions),
        ("嵌套推导式测试", demonstrate_nested_comprehensions),
        ("条件推导式测试", demonstrate_conditional_comprehensions),
        ("函数式推导式测试", demonstrate_functional_comprehensions),
        ("性能优化测试", demonstrate_performance_optimization),
    ]
    
    passed = 0
    total = len(test_functions)
    
    for test_name, test_func in test_functions:
        try:
            test_func()
            print(f"✓ {test_name} 通过")
            passed += 1
        except Exception as e:
            print(f"✗ {test_name} 失败: {e}")
            logger.exception(f"测试失败: {test_name}")
    
    # 运行异步测试
    try:
        asyncio.run(run_async_tests())
        print(f"✓ 异步推导式测试 通过")
        passed += 1
        total += 1
    except Exception as e:
        print(f"✗ 异步推导式测试 失败: {e}")
        logger.exception("异步测试失败")
        total += 1
    
    # 显示性能摘要
    summary = profiler.get_summary()
    if summary:
        print(f"\n✓ 性能摘要:")
        print(f"  总操作数: {summary['total_operations']}")
        print(f"  总耗时: {summary['total_time']:.4f}秒")
        print(f"  平均每操作耗时: {summary['avg_time_per_operation']:.6f}秒")
        print(f"  处理速度: {summary['avg_items_per_second']:.0f} 项/秒")
    
    print(f"\n测试结果: {passed}/{total} 通过")
    print("所有企业级推导式和集合操作最佳实践演示完成!")

if __name__ == "__main__":
    run_comprehensive_tests()